package com.dayuanit.atm.handler;

import com.dayuanit.atm.entity.User;
import com.dayuanit.atm.service.RedisService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

public class ATMWebSocketHandler extends TextWebSocketHandler implements InitializingBean {

    @Autowired
    private RedisService redisService;

    private static LinkedBlockingQueue<MessageHolder> queue = new LinkedBlockingQueue<>(1000);

    static {
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(">>>>>>>>>>>>>>>>发送消息的线程启动...");
                while (true) {
                    try {
                        MessageHolder messageHolder = queue.take();
                        System.out.println(">>>>>>>>>>>>>>>>从消息队列获取到消息 userid = " + messageHolder.getUserId());
                        final WebSocketSession webSocketSession = socketMap.get(messageHolder.getUserId());
                        if (null == webSocketSession) {
                            return;
                        }

                        try {
                            System.out.println(">>>>>>>>到账人=" +messageHolder.getUserId() + "，消息=" + messageHolder.getMessage());
                            webSocketSession.sendMessage(new TextMessage(messageHolder.getMessage()));
                        } catch (Exception e) {
                            e.printStackTrace();
                            socketMap.remove(messageHolder.getUserId());
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        t1.start();
    }

    @Override
    public void afterPropertiesSet() {
        System.out.println(">>>>>>>>>>>>>websocketHandler afer.......");
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    final MessageHolder messageHolder = redisService.popMessage();
                    System.out.println("redis队列获取信息发送");
                    System.out.println(">>>>>>>>到账人=" +messageHolder.getUserId() + "，消息=" + messageHolder.getMessage());
                    final WebSocketSession webSocketSession = socketMap.get(messageHolder.getUserId());
                    if (null == webSocketSession) {
                        return;
                    }
                    try {
                        System.out.println(">>>>>>>>到账人=" +messageHolder.getUserId() + "，消息=" + messageHolder.getMessage());
                        webSocketSession.sendMessage(new TextMessage(messageHolder.getMessage()));
                    } catch (Exception e) {
                        e.printStackTrace();
                        socketMap.remove(messageHolder.getUserId());
                    }
                }
            }
        }).start();
    }

    public static void sendMessage(MessageHolder messageHolder) {
        final WebSocketSession webSocketSession = socketMap.get(messageHolder.getUserId());
        if (null == webSocketSession) {
            return;
        }

        try {
            System.out.println(">>>>>>>>到账人=" +messageHolder.getUserId() + "，消息=" + messageHolder.getMessage());
            webSocketSession.sendMessage(new TextMessage(messageHolder.getMessage()));
        } catch (Exception e) {
            e.printStackTrace();
            socketMap.remove(messageHolder.getUserId());
        }
    }

    public void addMessage(int userId, String message) {
        final MessageHolder messageHolder = new MessageHolder();
        messageHolder.setMessage(message);
        messageHolder.setUserId(userId);

        boolean success = queue.offer(messageHolder);

    }

    //集合要线程安全的
    public static Map<Integer, WebSocketSession> socketMap = new ConcurrentHashMap<>();

    /**
     * 当websocket握手成功后 会调用此方法
     * @param session
     * @throws Exception
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        System.out.println(">>>>>>>>>>websocket建立成功");

        User user = (User)session.getAttributes().get("loginUser");
        socketMap.put(user.getId(), session);
    }

    /**
     * 客户端有消息过来 会调用此方法
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        System.out.println("获取客户端发送的消息=" + message.getPayload());

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(10 * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                try {
                    session.sendMessage(new TextMessage("我是服务器端的消息"));
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        }).start();
    }

    /**
     * 当连接断掉 会调用此方法
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        System.out.println("websocket连接断掉");

        User user = (User)session.getAttributes().get("loginUser");
        socketMap.remove(user.getId());
    }

}
